/*
 * PriorityPacketQueue.cpp
 *
 *  Created on: 2018年9月4日
 *      Author: zhengboyuan
 */

#include "PriorityPacketQueue.h"
#include "MediaType.h"


namespace av
{

PriorityPacketQueue::PriorityPacketQueue():
	m_waterLevel(),
	m_maxSize(MAX_QUEUE_SIZE)
{

}

PriorityPacketQueue::~PriorityPacketQueue()
{
}

void PriorityPacketQueue::setWaterLevel(int64_t duration)
{
	comn::AutoCritSec   lock(m_cs);
	m_waterLevel = duration;
}

void PriorityPacketQueue::setMaxSize(size_t maxSize)
{
	comn::AutoCritSec   lock(m_cs);
	m_maxSize = maxSize;
}

size_t PriorityPacketQueue::size()
{
	comn::AutoCritSec   lock(m_cs);
	return m_packets.size();
}

bool PriorityPacketQueue::empty()
{
	comn::AutoCritSec   lock(m_cs);
	return m_packets.empty();
}

size_t PriorityPacketQueue::push(AVPacketPtr& packet)
{
	size_t count = 0;
	int64_t duration = 0;
	bool isOverflow = false;
	{
		comn::AutoCritSec   lock(m_cs);
		m_packets.push_back(packet);

		count = m_packets.size();
		duration = getPtsDuration();
		isOverflow = (count >= m_maxSize || duration >= m_waterLevel);
	}

	if (isOverflow)
	{
		m_event.post();
	}
	return count;
}

AVPacketPtr PriorityPacketQueue::pop()
{
	AVPacketPtr packet;
	pop(packet);
	return packet;
}

bool PriorityPacketQueue::pop(AVPacketPtr& packet)
{
	bool found = false;
	comn::AutoCritSec   lock(m_cs);
	if (!m_packets.empty())
	{
		packet = m_packets.front();
		m_packets.pop_front();
		found = true;
	}
	return found;
}

void PriorityPacketQueue::clear()
{
	{
		comn::AutoCritSec   lock(m_cs);
		m_packets.clear();
	}

	m_event.post();
}

bool PriorityPacketQueue::timedwait(int ms)
{
	return m_event.timedwait(ms);
}

bool PriorityPacketQueue::pop(AVPacketPtr& packet, int ms)
{
	if (empty())
	{
		timedwait(ms);
	}
	return pop(packet);
}

bool PriorityPacketQueue::popSize(AVPacketPtr& packet, size_t& count)
{
	bool found = false;
	comn::AutoCritSec   lock(m_cs);
	count = m_packets.size();
	if (count > 0)
	{
		packet = m_packets.front();
		m_packets.pop_front();
		found = true;
		count -= 1;
	}
	return found;
}

size_t PriorityPacketQueue::popTimeout(AVPacketPtr& packet, int ms)
{
	size_t count = 0;
	if (popSize(packet, count))
	{
		return count;
	}

	if (!timedwait(ms))
	{
		return 0;
	}

	popSize(packet, count);
	return count;
}

void PriorityPacketQueue::cancelWait()
{
	m_event.post();
}

size_t PriorityPacketQueue::dropUntilKeyFrame()
{
	size_t count = 0;
	comn::AutoCritSec   lock(m_cs);

	size_t keyIdx = -1;
	for (size_t i = m_packets.size(); i > 0; --i)
	{
		size_t idx = i - 1;
		AVPacketPtr& packet = m_packets[idx];
		if (isVideoKey(packet))
		{
			keyIdx = idx;

			for (i = keyIdx; i > 0; --i)
			{
				idx = i - 1;
				AVPacketPtr& packet = m_packets[idx];
				if (isVideoKey(packet))
				{
					keyIdx = idx;
				}
				else
				{
					break;
				}
			}

			break;
		}
	}

	if (keyIdx == (size_t)-1)
	{
		return 0;
	}

	count = keyIdx;

	PacketDeque::iterator itBegin = m_packets.begin();
	PacketDeque::iterator it = m_packets.begin();
	std::advance(it, keyIdx);

	m_packets.erase(itBegin, it);

	return count;
}

bool PriorityPacketQueue::isVideoKey(AVPacketPtr& pkt)
{
	return ((pkt->flags & AV_PKT_FLAG_KEY) != 0) && (pkt->stream_index == av::MEDIA_TYPE_VIDEO);
}

size_t PriorityPacketQueue::insert(AVPacketPtr& packet)
{
	comn::AutoCritSec   lock(m_cs);

	PacketDeque::iterator pos = m_packets.end();
	PacketDeque::iterator it = m_packets.begin();
	for (; it != m_packets.end(); ++it)
	{
		if ((*it)->pts >= packet->pts)
		{
			pos = it;
			break;
		}
	}

	m_packets.insert(pos, packet);
	return m_packets.size();
}

int64_t PriorityPacketQueue::getDuration()
{
	comn::AutoCritSec   lock(m_cs);
	return getPtsDuration();
}

int64_t PriorityPacketQueue::getPosDuration()
{
	int64_t duration = 0;
	comn::AutoCritSec   lock(m_cs);
	if (m_packets.size() > 1)
	{
		duration = m_packets.back()->pos - m_packets.front()->pos;
	}
	return duration;
}

AVPacketPtr PriorityPacketQueue::front()
{
	AVPacketPtr pkt;
	comn::AutoCritSec   lock(m_cs);
	if (m_packets.size())
	{
		pkt = m_packets.front();
	}
	return pkt;
}

bool PriorityPacketQueue::getFrontPts(int64_t& pts)
{
	bool found = false;
	comn::AutoCritSec   lock(m_cs);
	if (m_packets.size())
	{
		pts = m_packets.front()->pts;
		found = true;
	}
	return found;
}

bool PriorityPacketQueue::getFrontPos(int64_t& pos)
{
	bool found = false;
	comn::AutoCritSec   lock(m_cs);
	if (m_packets.size())
	{
		pos = m_packets.front()->pos;
		found = true;
	}
	return found;
}

bool PriorityPacketQueue::popEx(AVPacketPtr& packet, int ms, int mediaType)
{
	for (int i = 0; i < 2; i ++)
	{
		{
			comn::AutoCritSec   lock(m_cs);

			int64_t duration = getPtsDuration();
			if (duration > m_waterLevel)
			{
				return popPacket(packet);
			}

			if (duration < 0)
			{
				return popPacket(packet);
			}

			if (m_packets.size() > m_maxSize)
			{
				return popPacket(packet);
			}

			if (m_packets.size() > 0 && mediaType >= 0)
			{
				if (m_packets.front()->stream_index == mediaType)
				{
					return popPacket(packet);
				}
			}
		}
		
		if (!m_event.timedwait(ms))
		{
			break;
		}
	}
	return false;
}

bool PriorityPacketQueue::popPacket(AVPacketPtr& packet)
{
	bool found = false;
	if (m_packets.size() > 0)
	{
		packet = m_packets.front();
		m_packets.pop_front();
		found = true;
	}
	return found;
}

int64_t PriorityPacketQueue::getPtsDuration()
{
	int64_t duration = 0;
	if (m_packets.size() > 1)
	{
		duration = m_packets.back()->pts - m_packets.front()->pts;
	}
	return duration;
}



} /* namespace av */
